package com.cuijian.demo.config;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

public class RocketMQConfig {

    // RocketMQ配置
    private String nameServer = "localhost:9876";
    private String producerGroup = "jfinal-producer-group";
    private String consumerGroup = "jfinal-consumer-group";

    private DefaultMQProducer producer;
    private DefaultMQPushConsumer consumer;
    private boolean mqEnabled = false; // MQ是否启用标志
    private static RocketMQConfig instance = new RocketMQConfig();

    private RocketMQConfig() {}

    public static RocketMQConfig getInstance() {
        return instance;
    }

    /**
     * 检查MQ是否可用
     */
    public boolean isMqEnabled() {
        return mqEnabled;
    }

    /**
     * 初始化生产者（带降级处理）
     */
    public void initProducer() {
        try {
            System.out.println("🔍 尝试初始化RocketMQ Producer...");
            System.out.println("📡 NameServer地址: " + nameServer);

            // 测试连接
            testMQConnection();

            producer = new DefaultMQProducer(producerGroup);
            producer.setNamesrvAddr(nameServer);
            producer.setVipChannelEnabled(false);
            producer.setRetryTimesWhenSendAsyncFailed(0);
            producer.setSendMsgTimeout(5000); // 设置发送超时时间

            producer.start();

            mqEnabled = true;
            System.out.println("✅ RocketMQ Producer启动成功");
            System.out.println("🎯 Producer Group: " + producerGroup);

        } catch (Exception e) {
            mqEnabled = false;
            System.out.println("⚠️ RocketMQ Producer启动失败，进入降级模式");
            System.out.println("❌ 错误信息: " + e.getMessage());
            System.out.println("ℹ️ 消息功能将使用本地队列处理或直接跳过");

            // 关闭可能部分初始化的生产者
            if (producer != null) {
                producer.shutdown();
                producer = null;
            }
        }
    }

    /**
     * 测试MQ连接
     */
    private void testMQConnection() {
        try {
            System.out.println("🔍 测试RocketMQ连接: " + nameServer);
            // 这里可以添加更复杂的连接测试逻辑
            // 例如尝试创建临时连接等

            // 简单的网络连接测试（可选）
            // InetAddress.getByName(nameServer.split(":")[0]).isReachable(3000);

        } catch (Exception e) {
            throw new RuntimeException("RocketMQ连接测试失败: " + e.getMessage(), e);
        }
    }

    /**
     * 初始化消费者（带降级处理）
     */
    public void initConsumer() {
        if (!mqEnabled) {
            System.out.println("⚠️ MQ未启用，跳过消费者初始化");
            return;
        }

        try {
            consumer = new DefaultMQPushConsumer(consumerGroup);
            consumer.setNamesrvAddr(nameServer);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            System.out.println("✅ RocketMQ Consumer初始化成功");
        } catch (Exception e) {
            System.out.println("⚠️ RocketMQ Consumer初始化失败: " + e.getMessage());
            consumer = null;
        }
    }

    /**
     * 获取生产者（降级模式返回null）
     */
    public DefaultMQProducer getProducer() {
        if (!mqEnabled) {
            return null;
        }
        if (producer == null) {
            initProducer();
        }
        return producer;
    }

    /**
     * 手动启用/禁用MQ
     */
    public void setMqEnabled(boolean enabled) {
        this.mqEnabled = enabled;
        if (!enabled) {
            shutdownProducer();
            shutdownConsumer();
            System.out.println("🔴 MQ已手动禁用");
        } else {
            System.out.println("🟢 MQ已手动启用");
            initProducer();
        }
    }

    /**
     * 重新连接MQ
     */
    public boolean reconnect() {
        System.out.println("🔄 尝试重新连接RocketMQ...");
        shutdownProducer();
        shutdownConsumer();

        try {
            initProducer();
            if (mqEnabled) {
                initConsumer();
                System.out.println("✅ RocketMQ重新连接成功");
                return true;
            }
        } catch (Exception e) {
            System.err.println("❌ RocketMQ重新连接失败: " + e.getMessage());
        }
        return false;
    }

    /**
     * 获取MQ状态信息
     */
    public String getMqStatus() {
        StringBuilder status = new StringBuilder();
        status.append("MQ状态: ").append(mqEnabled ? "🟢 已启用" : "🔴 已禁用").append("\n");
        status.append("NameServer: ").append(nameServer).append("\n");
        status.append("Producer状态: ").append(producer != null ? "🟢 运行中" : "🔴 未运行").append("\n");
        status.append("Consumer状态: ").append(consumer != null ? "🟢 运行中" : "🔴 未运行").append("\n");
        return status.toString();
    }

    /**
     * 关闭生产者
     */
    public void shutdownProducer() {
        if (producer != null) {
            try {
                producer.shutdown();
                System.out.println("✅ RocketMQ Producer已关闭");
            } catch (Exception e) {
                System.err.println("❌ 关闭Producer时出错: " + e.getMessage());
            } finally {
                producer = null;
            }
        }
    }

    /**
     * 关闭消费者
     */
    public void shutdownConsumer() {
        if (consumer != null) {
            try {
                consumer.shutdown();
                System.out.println("✅ RocketMQ Consumer已关闭");
            } catch (Exception e) {
                System.err.println("❌ 关闭Consumer时出错: " + e.getMessage());
            } finally {
                consumer = null;
            }
        }
    }

    /**
     * 应用停止时的清理
     */
    public void destroy() {
        System.out.println("🧹 清理RocketMQ资源...");
        shutdownProducer();
        shutdownConsumer();
        mqEnabled = false;
        System.out.println("✅ RocketMQ资源清理完成");
    }

    // Getter和Setter
    public String getNameServer() { return nameServer; }
    public void setNameServer(String nameServer) {
        this.nameServer = nameServer;
        // 修改地址后重新初始化
        if (producer != null) {
            shutdownProducer();
        }
        initProducer();
    }

    public String getProducerGroup() { return producerGroup; }
    public void setProducerGroup(String producerGroup) { this.producerGroup = producerGroup; }

    public String getConsumerGroup() { return consumerGroup; }
    public void setConsumerGroup(String consumerGroup) { this.consumerGroup = consumerGroup; }

    public DefaultMQProducer getProducerInstance() { return producer; }
    public DefaultMQPushConsumer getConsumerInstance() { return consumer; }
}